package JJJJJJava.dataframereflection;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.util.List;

/**
 * @Auther: Mengkunxuan
 * @Date:2018/9/2510:31
 * @Description:
 */
public class RDD2DateFrameReflection {
    public static void main(String[] args) {
        SparkConf conf  = new SparkConf().setMaster("local").setAppName("RDD2DataFrameReflection");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        final JavaRDD<String> lines = sc.textFile("c://");

        JavaRDD<Student> students = lines.map(new Function<String, Student>() {
            @Override
            public Student call(String s) throws Exception {
                 String [] stus = s.split(",");
                 Student stu = new Student();
                 stu.setId(Integer.valueOf(stus[0].trim()));
                stu.setName(stus[0].trim());
                stu.setAge(Integer.valueOf(stus[0].trim()));
                return stu;
            }
        });

        // 使用反射方式，将RDD转换为DataFrame
        // 将Student.class传入进去，其实就是用反射的方式来创建DataFrame
        // 因为Student.class本身就是反射的一个应用
        // 然后底层还得通过对Student Class进行反射，来获取其中的field
        // 这里要求，JavaBean必须实现Serializable接口，是可序列化的
        DataFrame studentDF = sqlContext.createDataFrame(students,Student.class);

        //拿到了一个DataFarme之后,就可以将其注册为一个临时表,然后针对其中的数据执行SQL语句
        studentDF.registerTempTable("studets");
        //针对students临时表执行SQL语句,查询年龄小于18的学生,就是teenageer
        DataFrame teenagerDF = sqlContext.sql("select * from students where age<=18");

        //将查询出来的DataFrame,再次转换为RDD
        final JavaRDD<Row> reenagerRDD = teenagerDF.javaRDD();

        //将RDD中的数据,进行映射,映射为student
        JavaRDD<Student> teenagerStudentRDD =reenagerRDD.map(new Function<Row, Student>() {
            @Override
            public Student call(Row row) throws Exception {
                //row中的数据的顺序,可能是跟我们期望的是不一样的!
                Student stu = new Student();
                stu.setAge(row.getInt(0));
                stu.setId(row.getInt(1));
                stu.setName(row.getString(2));
                return stu;
            }
        });

        //将数据collect回来,打印出来
        List<Student> studentList = teenagerStudentRDD.collect();
        for (Student stu:studentList)
        {
            System.out.println(stu);
        }
    }
}
